package com.example.pagerank;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

//第一阶段首先统计网页总数n，并将初始的每个网页的权值设置为 1/n
//输入文件中的每一行数据 即是M矩阵的每一列数据
/*
 Map阶段的任务如下：
输入：<源网页：所有链接网页网页>
输出：<链接网页：上一轮该链接网页的权值，链接网页的当前的部分权值>
 1）首先拿到缓存到的每个网页的权值数据，并将其放到map集合中
 2）读取某个源网页链接到的所有网页数量 从而获得到源网页到每个链接网页的跳转概率,
 3）再通过缓存数据得到源网页的权值 ，并将其参与计算 得出链接网页的部分权值 （其总权值需要reduce端合并）
 */
public class PageRankMapper extends
        Mapper<LongWritable, Text, Text, DoubleWritable> {
    Map<String, String> weightMap = new HashMap<String, String>();

    // 准备阶段将缓存进工作节点的概率分布表读出
    @Override
    protected void setup(Context context) throws IOException,
            InterruptedException {
        //拿到缓存文件数组
        URI[] uris = context.getCacheFiles();
        //往缓存文件上兑一根输入流
        BufferedReader br = new BufferedReader(new InputStreamReader(
                new FileInputStream(uris[0].getPath().toString())));

        String line;
        while (StringUtils.isNotEmpty(line = br.readLine())) {
            String field[] = line.split("\t");
            weightMap.put(field[0], field[1]);
        }

        br.close();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        String[] resource_links = line.split(":");
        String resource = resource_links[0];
        String[] links = resource_links[1].split("\t");
        boolean containRes = false;//记录是否存在链接到源网页自身的链接

        int linkNum = links.length;	//源网页中所有链接网页的数量
        double sourceWeight = Double.parseDouble(weightMap.get(resource));	//源网页权值
        //在源网页之中的所有链接网页的“当前”权值都一样
        double partLinkWeight = (1.0 / linkNum) * sourceWeight;

        for (int i = 0; i < linkNum; i++) {
            //当存在链接到源网页自身的链接时 布尔值改为true;
            if (links[i].equals(resource))
                containRes = true;

            // 键为  链接网页 ：上一轮链接网页的权值
            //这样做的目的是为了新、老权值的变化，从而判断迭代是否完成
            context.write(new Text(links[i] + ":" + weightMap.get(links[i])),
                    new DoubleWritable(partLinkWeight));
        }
        // 若网页的链接目标中没有本身，则输出<源网页地址,0.0> 以防丢失
        if (!containRes)
            context.write(
                    new Text(resource + ":"+ sourceWeight),
                    new DoubleWritable(0.0));
    }
}